package com.lsx143.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lsx143.realtime.app.BaseAppV2;
import com.lsx143.realtime.bean.OrderDetail;
import com.lsx143.realtime.bean.OrderInfo;
import com.lsx143.realtime.bean.OrderWide;
import com.lsx143.realtime.common.Constants;
import com.lsx143.realtime.util.DimUtil;
import com.lsx143.realtime.util.JdbcUtil;
import com.lsx143.realtime.util.RedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Map;

/**
 * DWM-订单维度宽表
 */
public class DWMOrderWideAppCache extends BaseAppV2 {
    public static void main(String[] args) {
        new DWMOrderWideAppCache().init(
                20005,
                "DWMOrderWideApp",
                1,
                "DWMOrderWideApp",
                Constants.TOPIC_DWD_ORDER_INFO,
                Constants.TOPIC_DWD_ORDER_DETAIL);
    }

    /**
     * 生成订单维度宽表的逻辑
     *
     * @param env        flink的执行环境
     * @param srcStreams 多个源数据流
     */
    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> srcStreams) {
        System.out.println("【DWMOrderWideAppCache】订单宽表App启动");
        //1.读取order_info和order_detail事实表
        SingleOutputStreamOperator<OrderWide> orderWideWithoutDimsStream = getOrderWideWithoutDims(srcStreams);

        //2.添加其他维度属性到宽表中
        SingleOutputStreamOperator<OrderWide> orderWideWithDimsStream = getOrderWideWithDims(orderWideWithoutDimsStream);
        orderWideWithDimsStream.print();
        //3.写出到kafka
    }

    /**
     * 将DIM维度表信息添加到宽表
     * 这种连接方式下有个弊端就是,没有一个订单都会去数据库查询,效率比较低
     *
     * @param orderWideWithoutDimsStream 订单与订单详情事实表join后的数据流
     * @return 订单数据宽表
     */
    private SingleOutputStreamOperator<OrderWide> getOrderWideWithDims(SingleOutputStreamOperator<OrderWide> orderWideWithoutDimsStream) {
        return orderWideWithoutDimsStream.map(new RichMapFunction<OrderWide, OrderWide>() {
            //连接对象
            private Connection phoenixCon;
            private Jedis jedis;

            //初始化jdbc连接到Phoenix
            @Override
            public void open(Configuration parameters) throws Exception {
                phoenixCon = JdbcUtil.getJdbcConnection(Constants.PHOENIX_DRIVER, Constants.PHOENIX_URL);
                jedis = RedisUtil.getRedisClient();
            }

            //关流
            @Override
            public void close() throws Exception {
                if (phoenixCon != null) {
                    phoenixCon.close();
                }
                if (jedis != null) {
                    jedis.close();
                }
            }

            @Override
            public OrderWide map(OrderWide orderWide) {

                //1.用户信息
                JSONObject userInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_USER_INFO, orderWide.getUser_id());
                orderWide.setUser_gender(userInfo.getString("GENDER"));
                orderWide.calcUserAgeByBirthday(userInfo.getString("BIRTHDAY"));

                //2.省份
                JSONObject baseProvince = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_PROVINCE, orderWide.getProvince_id());
                orderWide.setProvince_name(baseProvince.getString("NAME"));
                orderWide.setProvince_area_code(baseProvince.getString("AREA_CODE"));
                orderWide.setProvince_iso_code(baseProvince.getString("ISO_CODE"));
                orderWide.setProvince_3166_2_code(baseProvince.getString("ISO_3166_2"));

                //3.sku
                JSONObject skuInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_SKU_INFO, orderWide.getSku_id());
                orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                orderWide.setTm_id(skuInfo.getLong("TM_ID"));
                orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));
                orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                //4.spu
                JSONObject spuInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_SPU_INFO, orderWide.getSpu_id());
                orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                //5.目录
                JSONObject category3 = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_CATEGORY3, orderWide.getCategory3_id());
                orderWide.setCategory3_name(category3.getString("NAME"));

                //6.品牌
                JSONObject tm = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_TRADEMARK, orderWide.getTm_id());
                orderWide.setTm_name(tm.getString("TM_NAME"));

                return orderWide;
            }
        });
    }

    /**
     * 获取order_info和order_detail事实表
     *
     * @param srcStreams 输入数据流map
     * @return order_info和order_detail宽表
     */
    private SingleOutputStreamOperator<OrderWide> getOrderWideWithoutDims(Map<String, DataStreamSource<String>> srcStreams) {
        //1.读取两个流
        //  1.1 读取OrderInfo
        KeyedStream<OrderInfo, Long> keyedOrderInfoStream = srcStreams
                .get(Constants.TOPIC_DWD_ORDER_INFO)
                .map(json -> JSON.parseObject(json, OrderInfo.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((info, ts) -> info.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);


        //  1.2 读取OrderDetail
        KeyedStream<OrderDetail, Long> keyedOrderDetailStream = srcStreams
                .get(Constants.TOPIC_DWD_ORDER_DETAIL)
                .map(json -> JSON.parseObject(json, OrderDetail.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);
        //2.进行join
        return keyedOrderInfoStream
                .intervalJoin(keyedOrderDetailStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo info,
                                               OrderDetail detail,
                                               Context ctx,
                                               Collector<OrderWide> out) {
                        out.collect(new OrderWide(info, detail));
                    }
                });
    }
}
